package com.burges.net

import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment, _}

/**
  * 创建人    BurgessLee 
  * 创建时间   2020/1/25 
  * 描述
  */
object WorkCount {

	def main(args: Array[String]): Unit = {
		// 输入参数合法性检查
		val params = ParameterTool.fromArgs(args)

		//1.核定执行环境设定
		val evn = StreamExecutionEnvironment.getExecutionEnvironment
		// 2.指定数据源地址，读取输入数据
		val text: DataStream[String] = evn.readTextFile("E:\\ideaProject\\flink\\wordCountStreaming.txt")
		// 3.对数据集指定操作转换逻辑
		val counts: DataStream[(String, Int)] = text.flatMap(_.toLowerCase().split(" "))
				.filter(_.nonEmpty)
        		.map((_,1))
				.keyBy(0)
				.sum(1)
		// 4.规定计算结果输出位置
		if (params.has("output")) {
			counts.writeAsText(params.get("output"))
		} else{
			println("Printing result to stdout. Use --output to specify output path.")
			counts.print()
		}
		// 5.指定名称并处罚流式任务
		evn.execute("Streaming WordCount")
	}

}
